package com.rj.data.sink;

import io.searchbox.action.BulkableAction;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestResult;
import io.searchbox.core.Bulk;

import java.util.List;

public class ElasticSink implements Sink<List<BulkableAction>> {
    String indexName;
    JestClient jestClient;

    public ElasticSink(String indexName, JestClient jestClient) {
        this.indexName = indexName;
        this.jestClient = jestClient;
    }


    @Override
    public void writeLines(List<BulkableAction> actionList, boolean append, boolean force, String splitter) {
        try {
            Bulk.Builder builder = new Bulk.Builder()
                    .defaultIndex(indexName);
            builder.addAction(actionList);
            Bulk bulk = builder.build();
            JestResult jestResult = jestClient.execute(bulk);

            System.out.println(jestResult.isSucceeded());
            System.out.println(jestResult.getErrorMessage());
            actionList.clear();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
